package cn.admin.remoting.impl;

import cn.admin.remoting.*;
import cn.admin.remoting.locale.LogResources;
import cn.admin.remoting.util.LoggerInit;
import org.apache.commons.lang.StringUtils;
import org.apache.mina.common.IoAcceptor;
import org.apache.mina.common.ThreadModel;
import org.apache.mina.transport.socket.nio.SocketAcceptor;
import org.apache.mina.transport.socket.nio.SocketAcceptorConfig;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.Executor;

import static cn.admin.remoting.locale.LogResources.SERVER_EMPTY_GROUP;

public class DefaultServer extends DefaultConnectionFactory implements Server {
    private static final int BACKLOG_SIZE = 1024;
    private final InetSocketAddress serversocket;
    private final IoAcceptor acceptor;
    private volatile boolean started;
    private volatile boolean stopped;
    private final Random random;
    private volatile GroupConnectionStrategy connSelector;

    DefaultServer(String bindIp, int port) {
        this.started = false;
        this.started = false;
        this.random = new Random();
        this.connSelector = new RandomConnectionStrategy();
        if (StringUtils.isBlank(bindIp)) {
            this.serversocket = new InetSocketAddress(port);
        } else {
            this.serversocket = new InetSocketAddress(bindIp,port);
        }
        this.acceptor = new SocketAcceptor(processorCount,IO_THREAD_POOL);
        SocketAcceptorConfig cfg = ((SocketAcceptor) this.acceptor).getDefaultConfig();
        cfg.setBacklog(BACKLOG_SIZE);
        cfg.setThreadModel(ThreadModel.MANUAL);
        cfg.getFilterChain().addLast(CODEC_FILTER_NAME,CODEC_FILTER);
    }

    private Connection randomGetConnectionByGroup(String group) {

        Collection<Connection> connections = this.groupConnectionSnapshot(group);
        if (connections.isEmpty()) {
            return null;
        } else if (connections.size() == 1) {
            return connections.iterator().next();
        } else {
            Connection[] connectionsArray = (Connection[]) connections.toArray();
            return connectionsArray[this.random.nextInt(connections.size())];
        }

    }

    @Override
    public int getPort() {
        return this.serversocket.getPort();
    }

    @Override
    public String getBindIp() {
        return this.serversocket.getAddress().getHostAddress();
    }

    @Override
    public synchronized void start() throws RemotingException {
        if (this.stopped) {
            LoggerInit.LOGGER.error("不能启动已经关闭的DefaultServer");
        }

        try {
            if (!this.started) {
                DefaultIoHandler ioHandler = new DefaultIoHandler(this);
                this.acceptor.bind(this.serversocket, ioHandler);
                this.started = true;
                LoggerInit.LOGGER.warn("服务器已启动：" + this.serversocket);
            }
        } catch (IOException e) {
            throw new RemotingException("启动Server失败",e);
        }

        Runtime.getRuntime().addShutdownHook(new Thread(DefaultServer.this::stop));
    }

    @Override
    public void stop() {
        this.started = false;
        this.stopped = true;
        this.acceptor.unbindAll();
        LoggerInit.LOGGER.warn("服务器已停止运行：" + this.serversocket);
    }

    @Override
    public boolean isStarted() {
        return this.started;
    }

    public boolean isStopped() {
        return this.stopped;
    }

    @Override
    public void sendToGroups(Map<String, Object> groupDatas,
                             ResponseCallback.GroupRandomResponseCallback callback, long timeout,
                             Object... callbackArgs) throws RemotingException {
        final Map<String,ResponseOfRequest> responses = new HashMap<>();
        for (Map.Entry<String,Object> entry : groupDatas.entrySet()) {
            final String group = entry.getKey();
            Object data = entry.getValue();
            Connection connection = this.connSelector.getConnection(group);
            final String remoteIp = connection != null ? connection.getRemoteAddress() : null;
            final int remotePort = connection != null ? connection.getRemotePort() : -1;
            ResponseCallback innerCallback = new ResponseCallback() {
                @Override
                public Executor getExecutor() {
                    return callback.getExecutor();
                }

                @Override
                public void handleResponse(Object appResp) {
                    this.onGroupResponse(new ResponseOfRequest(appResp,remoteIp,remotePort));
                }

                @Override
                public void onRemotingException(int errorType, String errorMsg) {
                    this.onGroupResponse(new ResponseOfRequest(errorType,errorMsg,remoteIp,remotePort));
                }

                private void onGroupResponse(ResponseOfRequest groupResp) {
                    Map<String,ResponseOfRequest> responses = new HashMap<>();
                    synchronized (this) {
                        responses.put(group,groupResp);
                        if (responses.size() == groupDatas.size()) {
                            callback.handleResponses(responses,callbackArgs);
                        }
                    }
                }
            };

            if (connection != null) {
                RequestControl control = new RequestControlImpl((byte) 1,timeout);
                connection.getClient().invokeWithCallback(data,innerCallback,control);
            } else {
                ConnectionResponse response = new ConnectionResponse();
                response.setResult(3);
                innerCallback.onRemotingException(3, LogResources.getLog(SERVER_EMPTY_GROUP,
                        group));
            }
        }

    }

    @Override
    public void sendToGroupOneway(String group, Object payloadObject) throws RemotingException {
        Connection connection = this.connSelector.getConnection(group);
        if (connection == null) {
            throw new RemotingException(LogResources.getLog(SERVER_EMPTY_GROUP,group));
        } else {
            RequestControl control = new RequestControlImpl((byte)1);
            connection.getClient().oneway(payloadObject,control);
        }
    }

    @Override
    public ResponseFuture invokeWithFuture(Object appRequest, Connection connection,
                                           long waitTimeoutMs, byte protocol)
            throws RemotingException {
        RequestControl control = new RequestControlImpl(protocol,waitTimeoutMs);
        return connection.getClient().invokeWithFuture(appRequest,control);
    }

    @Override
    public void sendToGroupAllOneway(String group, Object payload, byte protocol) {
        RequestControl control = new RequestControlImpl(protocol);
        for (Connection conn : this.groupConnectionSnapshot(group)) {
            conn.getClient().oneway(payload,control);
        }
    }

    @Override
    public void sendToGroupAllCallback(String group, Object payload,
                                       ResponseCallback.MultiResponseCallback totalCallback,
                                       byte protocol, long waitTimeoutMs) {
        Collection<Connection> conns = this.groupConnectionSnapshot(group);
        ConnectionsWithOneCallback help = new ConnectionsWithOneCallback(conns,totalCallback);

        for (Connection connection : conns) {
            RequestControl control = new RequestControlImpl(protocol,waitTimeoutMs);
            connection.getClient().invokeWithCallback(payload,help.getConnectionCallback(connection),
                    control);
        }
    }

    @Override
    public void setGroupConnectionStrategy(GroupConnectionStrategy strategy) {
        if (strategy == null) {
            throw new IllegalArgumentException();
        } else {
            this.connSelector = strategy;
        }
    }

    private class RandomConnectionStrategy implements GroupConnectionStrategy {

        @Override
        public Connection getConnection(String groupName) {
            return DefaultServer.this.randomGetConnectionByGroup(groupName);
        }
    }

    static class ConnectionsWithOneCallback {

        final Collection<Connection> connections;
        final ResponseCallback.MultiResponseCallback totalCallback;
        final Map<Connection,ResponseCallback> conn2callback;
        final Map<Connection,ResponseOfRequest> appResponses;

        ConnectionsWithOneCallback(Collection<Connection> connections,
                                   ResponseCallback.MultiResponseCallback totalCallback) {
            this.connections = connections;
            this.totalCallback = totalCallback;
            this.conn2callback = new HashMap<>();
            this.appResponses = new HashMap<>();

            for (Connection connection : connections) {
                this.conn2callback.put(connection, new ResponseCallback() {
                    @Override
                    public Executor getExecutor() {
                        return ConnectionsWithOneCallback.this.totalCallback.getExecutor();
                    }

                    @Override
                    public void handleResponse(Object appResp) {
                        this.onCallback(new ResponseOfRequest(appResp,connection.getRemoteAddress(),
                                connection.getRemotePort()));
                    }

                    @Override
                    public void onRemotingException(int errorType, String errorMsg) {
                        this.onCallback(new ResponseOfRequest(errorType,errorMsg,connection.getRemoteAddress(),
                                connection.getRemotePort()));
                    }

                    private void onCallback(ResponseOfRequest resp) {
                        synchronized (ConnectionsWithOneCallback.this.appResponses) {
                            ConnectionsWithOneCallback.this.appResponses.put(connection,resp);
                            if (ConnectionsWithOneCallback.this.appResponses.size() ==
                                    ConnectionsWithOneCallback.this.connections.size()) {
                                ConnectionsWithOneCallback.this.totalCallback.onMultiResponse(new
                                        ArrayList<>(ConnectionsWithOneCallback.
                                        this.appResponses.values()));
                            }
                        }
                    }
                });
            }
        }

        ResponseCallback getConnectionCallback(Connection connection) {
            return this.conn2callback.get(connection);
        }

    }

}
